/*实现异步工作器*/
#pragma once

#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace nhylog
{

    // 回调类型
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        //安全状态标识缓冲区满了阻塞，避免资源耗尽
        ASYNC_SAFE,
        //unsafe不考虑资源耗尽，无限扩容，压力测试
        ASYNC_UNSAFE
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor& cb,AsyncType looper_type = AsyncType::ASYNC_SAFE)
            : _stop(false)
            , _thread(std::thread(&AsyncLooper::threadEntry, this))
            , _callBack(cb)
            , _looper_type(looper_type)
        {
        }

        ~AsyncLooper()
        {
            stop();
        }

        void stop()
        {
            _stop = true;

            // 唤醒所有工作线程
            _cond_con.notify_all();
            
            //等待所有工作线程结束后再退出
            _thread.join();
        }

        void push(const char *data, size_t len)
        {
            // 1 无限扩容-非安全
            // 2.固定大小-生产缓冲区数据满-阻塞

            std::unique_lock<std::mutex> lock(_mutex);
            // 若缓冲区剩余空间大小大于数据长度，返回真
            if(_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&](){ return _pro_buf.writeAbleSize() >= len; });

            // 走到这说明满足条件，可以向缓冲区添加数据

            _pro_buf.push(data, len);
            // 唤醒消费者，对缓冲区数据处理
            _cond_con.notify_all();
        }

    private:
        void threadEntry()
        {
            while (true)
            {
                // 1、判断生产缓冲区有没有数据，有交换、没阻塞

                // 为互斥锁设置一个声明周期，不对处理数据加锁保护
                {
                    // 交换完后解锁即可
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 退出前被唤醒，或有数据被唤醒，继续往下走，否则休眠
                    _cond_con.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });
                    if(_stop && _pro_buf.empty()) break;
                    _con_buf.swap(_pro_buf);

                    // 2.唤醒生产者
                    _cond_pro.notify_all();
                }

                // 3、被唤醒后，对消费缓冲区进行数据处理，只有一个处理线程，不需要加锁
                _callBack(_con_buf);

                // 4、初始化消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack;

    private:
        // 工作器停止标志
        std::atomic<bool> _stop;

        Buffer _pro_buf; // 生产缓冲区
        Buffer _con_buf; // 消费缓冲区

        std::mutex _mutex;
        // 两个条件变量
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;

        std::thread _thread;

        AsyncType _looper_type;
    };
}